package com.me.videoservice.config;

import com.me.videoservice.mqtt.MqttMessageHandler;
import javax.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

@Configuration
@Slf4j
public class MqttConfig {

  @Resource
  private MqttProperties mqttProperties;

  @Bean
  public MqttConnectOptions getMqttConnectOptions() {
    MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
    mqttConnectOptions.setCleanSession(true);
    mqttConnectOptions.setAutomaticReconnect(true);
    mqttConnectOptions.setConnectionTimeout(mqttProperties.getTimeout());
    mqttConnectOptions.setKeepAliveInterval(mqttProperties.getKeepalive());
    mqttConnectOptions.setUserName(mqttProperties.getUsername());
    mqttConnectOptions.setPassword(mqttProperties.getPassword().toCharArray());
    mqttConnectOptions.setServerURIs(mqttProperties.getBrokers());
    return mqttConnectOptions;
  }

  @Bean
  public MqttPahoClientFactory mqttClientFactory() {
    DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
    factory.setConnectionOptions(getMqttConnectOptions());
    return factory;
  }

  @Bean
  public MessageChannel mqttInputChannel() {
    return new DirectChannel();
  }

  @Bean
  public MessageChannel mqttOutputChannel() {
    return new DirectChannel();
  }


  @Bean
  public MessageProducer inbound() {
    MqttPahoMessageDrivenChannelAdapter adapter =
        new MqttPahoMessageDrivenChannelAdapter(mqttProperties.getClientId() + "/receive",
            mqttClientFactory(),
            mqttProperties.getTopics());
    adapter.setConverter(new DefaultPahoMessageConverter());
    adapter.setQos(1);
    adapter.setOutputChannel(mqttInputChannel());
    return adapter;
  }

  @ServiceActivator(inputChannel = "mqttInputChannel")
  @Bean
  public MessageHandler handler() {
    return new MqttMessageHandler();
  }

  @Bean
  @ServiceActivator(inputChannel = "mqttOutputChannel")
  public MessageHandler mqttOutbound(MqttPahoClientFactory mqttClientFactory) {
    MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(
        mqttProperties.getClientId() + "/send", mqttClientFactory);
    messageHandler.setAsync(true);
    messageHandler.setDefaultQos(1);
    messageHandler.setDefaultRetained(false);
    messageHandler.setAsyncEvents(false);
    return messageHandler;
  }

}
